/*
 * LimitedRateStreamingSource.java
 *
 * Copyright 锟�998-2010 Research In Motion Ltd.
 * 
 * Note: For the sake of simplicity, this sample application may not leverage
 * resource bundles and resource strings.  However, it is STRONGLY recommended
 * that application developers make use of the localization features available
 * within the BlackBerry development platform to ensure a seamless application
 * experience across a variety of languages and geographies.  For more information
 * on localizing your application, please refer to the BlackBerry Java Development
 * Environment Development Guide associated with this release.
 */

package fm.douban;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Random;

import javax.microedition.io.Connector;
import javax.microedition.io.ContentConnection;
import javax.microedition.media.Control;
import javax.microedition.media.protocol.ContentDescriptor;
import javax.microedition.media.protocol.DataSource;
import javax.microedition.media.protocol.SourceStream;

/**
 * The data source used by the BufferedPlayback's media player.
 */
public final class LimitedRateStreamingSource extends DataSource {
	/** The max size to be read from the stream at one time */
	private static final int READ_CHUNK = 512; // bytes
	private int DEBUG_LEVE = 0;
	/** A reference to the field which displays the load status */
	// private TextField _loadStatusField;

	/** A reference to the field which displays the player status */
	// private TextField _playStatusField;

	/**
	 * The minimum number of bytes that must be buffered before the media file
	 * will begin playing.
	 */
	private int _startBuffer = 20000;

	/** The maximum size (in bytes) of a single read */
	private int _readLimit = 32000;

	/**
	 * The minimum forward byte buffer which must be maintained in order for the
	 * video to keep playing. If the forward buffer falls below this number, the
	 * playback will pause until the buffer increases.
	 */
	private int _pauseBytes = 64000;

	/**
	 * The minimum forward byte buffer required to resume playback after a
	 * pause.
	 */
	private int _resumeBytes = 128000;

	/** The stream connection over which media content is passed */
	private ContentConnection _contentConnection;

	/** An input stream shared between several readers */
	private InputStream _readAhead;

	/** A stream to the buffered resource */
	private LimitedRateSourceStream _feedToPlayer;

	/** The MIME type of the remote media file */
	private String _forcedContentType;

	/** A counter for the total number of buffered bytes */
	private volatile int _totalRead;

	/** A flag used to tell the connection thread to stop */
	private volatile boolean _stop;

	/**
	 * A flag used to indicate that the initial buffering is complete. In other
	 * words, that the current buffer is larger than the defined start buffer
	 * size.
	 */
	private volatile boolean _bufferingComplete;

	/** A flag used to indicate that the remote file download is complete */
	private volatile boolean _downloadComplete;

	/** The thread which retrieves the remote media file */
	private ConnectionThread _loaderThread;

	private CircularByteBuffer buffer;
	private InputStream in;
	private OutputStream out;

	private long fileSize = 0;
	private boolean _downFalied = false;

	// private Random rand = new Random();

	/**
	 * Constructor
	 * 
	 * @param locator
	 *            The locator that describes the DataSource.
	 */
	LimitedRateStreamingSource(String locator) {
		super(locator);
		buffer = new CircularByteBuffer(1024 * 2);
		in = buffer.getInputStream();
		out = buffer.getOutputStream();
	}

	/**
	 * Open a connection to the locator
	 * 
	 * @throws IOException
	 *             Thrown if the firewall disallows a connection that is not
	 *             btspp or comm or if save file could not be created
	 */
	public void connect() throws IOException {
		// Open the connection to the remote file
		System.out.println("in connect():" + getLocator());
		_contentConnection = (ContentConnection) Connector.open(getLocator()
				+ Radio.getConnectType(), Connector.READ_WRITE);
		// Cache a reference to the locator
		String locator = getLocator();
		// Report status
		System.out.println("Loading: " + locator);
		System.out.println("Size: " + _contentConnection.getLength());
		fileSize = _contentConnection.getLength();
		if(fileSize < 1000000)
		{
			buffer = new CircularByteBuffer(16000);
			in = buffer.getInputStream();
			out = buffer.getOutputStream();
		}
		// Use a new shared input stream for buffered reading
		_readAhead = _contentConnection.openInputStream();
		if (_forcedContentType != null) {
			// Use the user-defined content type if it is set

			_feedToPlayer = new LimitedRateSourceStream(in, _forcedContentType);
		} else {
			// Otherwise, use the MIME types of the remote file
			_feedToPlayer = new LimitedRateSourceStream(in, _contentConnection
					.getType());
		}
	}

	/**
	 * Destroy and close all existing connections
	 */
	public void disconnect() {
		try {

			if (_readAhead != null) {
				// Close the reader stream
				_readAhead.close();
				_readAhead = null;
			}

			// Close the remote file connection
			_contentConnection.close();

			// Close the stream to the player
			_feedToPlayer.close();

			_loaderThread = null;
		} catch (Exception e) {
			Radio.errorDialog(e.getMessage());
		}
	}

	/**
	 * Returns the content type of the remote file
	 * 
	 * @return The content type of the remote file
	 */
	public String getContentType() {
		return _feedToPlayer.getContentDescriptor().getContentType();
	}

	/**
	 * Returns a stream to the buffered resource
	 * 
	 * @return A stream to the buffered resource
	 */
	public SourceStream[] getStreams() {
		return new SourceStream[] { _feedToPlayer };
	}

	/**
	 * Starts the connection thread used to download the remote file
	 */
	public void start() throws IOException {
		// If the save stream is null, we have already completely downloaded
		// the file.
		if (_readAhead != null) {
			// Open the connection thread to finish downloading the file
			_loaderThread = new ConnectionThread();
			_loaderThread.start();
		}
	}

	/**
	 * Stop the connection thread
	 */
	public void stop() throws IOException {
		// Set the boolean flag to stop the thread
		_stop = true;
	}

	/**
	 * @see javax.microedition.media.Controllable#getControl(String)
	 */
	public Control getControl(String controlType) {
		// No implemented Controls
		return null;
	}

	/**
	 * @see javax.microedition.media.Controllable#getControls()
	 */
	public Control[] getControls() {
		// No implemented Controls
		return null;
	}

	/**
	 * Force the lower level stream to a given content type. Must be called
	 * before the connect function in order to work.
	 * 
	 * @param contentType
	 *            The content type to use.
	 */
	void setContentType(String contentType) {
		_forcedContentType = contentType;
	}

	/**
	 * A stream to the buffered media resource
	 */
	private final class LimitedRateSourceStream implements SourceStream {
		/** A stream to the local copy of the remote resource */
		private InputStream _baseSharedStream;
		private long total;
		/** Describes the content type of the media file */
		private ContentDescriptor _contentDescriptor;

		/**
		 * Constructor. Creates a LimitedRateSourceStream from the given
		 * InputStream.
		 * 
		 * @param inputStream
		 *            The input stream used to create a new reader.
		 * @param contentType
		 *            The content type of the remote file.
		 */
		LimitedRateSourceStream(InputStream inputStream, String contentType) {
			this._baseSharedStream = inputStream;
			_contentDescriptor = new ContentDescriptor(contentType);
		}

		/**
		 * Returns the content descriptor for this stream
		 * 
		 * @return The content descriptor for this stream
		 */
		public ContentDescriptor getContentDescriptor() {
			return _contentDescriptor;
		}

		/**
		 * Returns the length provided by the connection
		 * 
		 * @return long The length provided by the connection
		 */
		public long getContentLength() {
			return _contentConnection.getLength();
		}

		/**
		 * Returns the seek type of the stream
		 */
		public int getSeekType() {
			return SEEKABLE_TO_START;
		}

		/**
		 * Returns the maximum size (in bytes) of a single read
		 */
		public int getTransferSize() {
			return _readLimit;
		}

		/**
		 * Writes bytes from the buffer into a byte array for playback
		 * 
		 * @param bytes
		 *            The buffer into which the data is read
		 * @param off
		 *            The start offset in array b at which the data is written
		 * @param len
		 *            The maximum number of bytes to read
		 * @return the total number of bytes read into the buffer, or -1 if
		 *         there is no more data because the end of the stream has been
		 *         reached.
		 * @throws IOException
		 *             Thrown if a read error occurs
		 */
		public int read(byte[] bytes, int off, int len) throws IOException {
			// System.out.println("Read Request for: " + len + " bytes");
			// System.out.println("Read Request start from : " + off +
			// " bytes");
			int ret = in.read(bytes, off, len);
			total = total + ret;
			// System.out.println("total:" + total + " buffer availiable:"
			// + buffer.getAvailable());
			long bytesToReturn = Math.min((fileSize - total), len);
			// System.out.println("bytesToReturn:" + bytesToReturn);
			if (bytesToReturn < 1 || _downFalied) {
				return -1;
			}
			return ret;
			// Limit bytes read to the readLimit
			/*
			 * int readLength = len; if (readLength > getReadLimit()) {
			 * readLength = getReadLimit(); }
			 * 
			 * long bytesToReturn = Math.min((fileSize - total), len);
			 * System.out.println("bytesToReturn:" + bytesToReturn); if
			 * (bytesToReturn < 1) { return -1; } // The number of available
			 * byes in the buffer int available;
			 * 
			 * // A boolean flag indicating that the thread should pause //
			 * until the buffer has increased sufficiently. boolean paused =
			 * false; int ret = in.read(bytes, off, len);
			 * System.out.println("Read1:" + ret); total = total + ret; return
			 * ret; for (int i = 0;; i++) {
			 * 
			 * // available = _baseSharedStream.available(); available =
			 * buffer.getAvailable(); if (DEBUG_LEVE == 1) {
			 * System.out.println("available:" + available); System.out
			 * .println("_downloadComplete in _baseSharedStream:" +
			 * _downloadComplete); System.out
			 * .println("_bufferingComplete in _baseSharedStream:" +
			 * _bufferingComplete);
			 * System.out.println("paused in _baseSharedStream:" + paused);
			 * System.out.println("getResumeBytes() in _baseSharedStream:" +
			 * getResumeBytes());
			 * System.out.println("readLength in _baseSharedStream:" +
			 * readLength); } if (_downloadComplete) { // Ignore all
			 * restrictions if downloading is complete
			 * System.out.println("Complete, Reading: " + len + " - Available: "
			 * + available); int res = _baseSharedStream.read(bytes, off, len);
			 * System.out.println("Read1:" + res); total = total + res;
			 * System.out.println("filesize:" + fileSize + " total read:" +
			 * total); return res; } else if (_bufferingComplete) { if (paused
			 * && available > getResumeBytes()) { // If the video is paused due
			 * to buffering, but the // number of available byes is sufficiently
			 * high, // resume playback of the media. System.out
			 * .println("Resuming - Available: " + available); paused = false;
			 * // total = total + readLength; int res =
			 * _baseSharedStream.read(bytes, off + readLength * i, readLength);
			 * System.out.println("Read2:" + res); total = total + res;
			 * System.out.println("filesize:" + fileSize + " total read:" +
			 * total); return res; } else if (!paused && (available >
			 * getPauseBytes() || available > readLength)) { // We have enough
			 * information for this media playback
			 * 
			 * if (available < getPauseBytes()) { // If the buffer is now
			 * insufficient, set the // pause flag. paused = true; }
			 * 
			 * System.out.println("Reading: " + readLength + " - Available: " +
			 * available); int res = _baseSharedStream.read(bytes, off +
			 * readLength * i, readLength); System.out.println("Read3:" + res);
			 * total = total + res; System.out.println("filesize:" + fileSize +
			 * " total read:" + total); return res;// return // res; } else if
			 * (!paused) { // Set pause until loaded enough to resume paused =
			 * true; } } else { // We are not ready to start yet, try sleeping
			 * to allow the // buffer to increase. try { Thread.sleep(500); }
			 * catch (Exception e) {
			 * Radio.errorDialog("Thread.sleep(long) threw " + e.getMessage());
			 * } } }
			 */
		}

		/**
		 * @see javax.microedition.media.protocol.SourceStream#seek(long)
		 */
		public long seek(long where) throws IOException {
			// _baseSharedStream.setCurrentPosition((int) where);
			System.out.println("seeking:==========where:" + where + " total:"
					+ total);
			return total;
		}

		/**
		 * @see javax.microedition.media.protocol.SourceStream#tell()
		 */
		public long tell() {
			System.out.println("telling:==========total:" + total);
			return total;
		}

		/**
		 * Close the stream
		 * 
		 * @throws IOException
		 *             Thrown if the stream could not be closed
		 */
		void close() throws IOException {
			_baseSharedStream.close();
		}

		/**
		 * @see javax.microedition.media.Controllable#getControl(String)
		 */
		public Control getControl(String controlType) {
			// No implemented controls
			return null;
		}

		/**
		 * @see javax.microedition.media.Controllable#getControls()
		 */
		public Control[] getControls() {
			// No implemented controls
			return null;
		}
	}

	/**
	 * A thread which downloads the remote file and writes it to the local file
	 */
	private final class ConnectionThread extends Thread {

		private Random rand = new Random();

		/**
		 * Download the remote media file, then write it to the local file.
		 * 
		 * @see java.lang.Thread#run()
		 */
		public void run() {
			try {
				byte[] data = new byte[READ_CHUNK];
				int len = 0;

				// Until we reach the end of the file
				while (_readAhead != null
						&& -1 != (len = _readAhead.read(data))) {
					_totalRead += len;

					if (!_bufferingComplete && _totalRead > getStartBuffer()) {
						// We have enough of a buffer to begin playback
						_bufferingComplete = true;
						System.out.println("Initial Buffering Complete");
					}
					if (_stop) {
						// Stop reading
						return;
					}
					out.write(data);
					// System.out.println("buffer:" + buffer.getAvailable());
					// System.out.println("_totalRead:" + _totalRead);
					// Thread.sleep(rand.nextInt(80));
					// Thread.yield();
					if (DEBUG_LEVE == 1) {
						System.out.println("_readAhead:" + _readAhead);
						System.out.println("_bufferingComplete:"
								+ _bufferingComplete);
						System.out.println("getStartBuffer():"
								+ getStartBuffer());
						System.out.println("_stop:" + _stop);
					}
				}

				System.out.println("Downloading Complete");
				// updateLoadStatus("Done " + _totalRead + " Bytes");
				System.out.println("Total Read: " + _totalRead);

				// If the downloaded data is not the same size
				// as the remote file, something is wrong.
				if (_totalRead != _contentConnection.getLength()) {
					System.err.println("* Unable to Download entire file *");
				}

				_downloadComplete = true;

			} catch (IOException e) {
				System.out.println("in download e:" + e);
				System.out.println("in download e:" + e.getMessage());
				_downFalied = true;
				// } catch (InterruptedException e) {
				// System.out.println("in download e:" + e);
				// System.out.println("in download e:" + e.getMessage());
				//
			}

		}
	}

	/**
	 * Gets the minimum forward byte buffer which must be maintained in order
	 * for the video to keep playing.
	 * 
	 * @return The pause byte buffer.
	 */
	int getPauseBytes() {
		return _pauseBytes;
	}

	/**
	 * Sets the minimum forward buffer which must be maintained in order for the
	 * video to keep playing.
	 * 
	 * @param pauseBytes
	 *            The new pause byte buffer
	 */
	void setPauseBytes(int pauseBytes) {
		_pauseBytes = pauseBytes;
	}

	/**
	 * Gets the maximum size (in bytes) of a single read
	 * 
	 * @return The maximum size (in bytes) of a single read
	 */
	int getReadLimit() {
		return _readLimit;
	}

	/**
	 * Sets the maximum size (in bytes) of a single read
	 * 
	 * @param readLimit
	 *            The new maximum size (in bytes) of a single read
	 */
	void setReadLimit(int readLimit) {
		_readLimit = readLimit;
	}

	/**
	 * Gets the minimum forward byte buffer required to resume playback after a
	 * pause.
	 * 
	 * @return The resume byte buffer
	 */
	int getResumeBytes() {
		return _resumeBytes;
	}

	/**
	 * Sets the minimum forward byte buffer required to resume playback after a
	 * pause.
	 * 
	 * @param resumeBytes
	 *            The new resume byte buffer
	 */
	void setResumeBytes(int resumeBytes) {
		_resumeBytes = resumeBytes;
	}

	/**
	 * Gets the minimum number of bytes that must be buffered before the media
	 * file will begin playing.
	 * 
	 * @return The start byte buffer
	 */
	int getStartBuffer() {
		return _startBuffer;
	}

	/**
	 * Sets the minimum number of bytes that must be buffered before the media
	 * file will begin playing.
	 * 
	 * @param startBuffer
	 *            The new start byte buffer
	 */
	void setStartBuffer(int startBuffer) {
		_startBuffer = startBuffer;
	}

}